import org.apache.spark.streaming.twitter.TwitterUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Twitter {
  def main(args: Array[String]): Unit = {
    val consumerKey = "75idq5t4XC6O866iyxToPmB8N"
    val consumerSecret = "J5AZOS0opyD5qoFvPkhC08QAtLpmjv9sDXIIDit8D9IUrRO7U6"
    val accessToken = "983580846316662784-s1b8Tnix5sALKdXFfXiiEzs1ugT1CTH"
    val accessTokenSecret = "geQG5YZaCOaYql1kCeddtrFzhFvAGwcFFsCtGvKWTzbpa"

    val arr = Array(consumerKey,consumerSecret,accessToken,accessTokenSecret,"jp","5","1","100")

    val sparkConf = new SparkConf().setAppName("streamingPratice")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(5))
    ssc.checkpoint("checkpoint")

    val tweets = TwitterUtils.createStream(ssc,None,arr)
    val tweetsFilteredByLang = tweets.filter{tweet => tweet.getLang == "jp"}
    val statuses = tweetsFilteredByLang.map{tweet => tweet.getText}
    val words = statuses.flatMap{status => status.split("""\s+""")}
    val hashTags = words.filter{word => word.startsWith("#")}
    val hashTagPairs = hashTags.map{hashtag => (hashtag , 1)}
    val tagsWithCounts = hashTagPairs.updateStateByKey(
      (counts: Seq[Int], prevCount: Option[Int]) =>
        prevCount.map{c => c + counts.sum}.orElse{Some(counts.sum)}
    )
    val topHashTags = tagsWithCounts.filter{case(t,c) => c > 1.toInt}
    val sortedTopHashTags = topHashTags.transform{
      rdd => rdd.sortBy({case(w,c) => c}, false)
    }
    sortedTopHashTags.print(100.toInt)
    ssc.start()
    ssc.awaitTermination()

  }
}
